CTA 3.0

Contents

CTA 3.0#

import warnings

import pandas as pd
import numpy as np

from cvx.simulator import Portfolio
from cvx.simulator import interpolate

warnings.simplefilter(action="ignore", category=FutureWarning)

# Load prices
prices = pd.read_csv("data/Prices_hashed.csv", index_col=0, parse_dates=True)

# interpolate the prices
prices = prices.apply(interpolate)

We use the system: $\(\mathrm{CashPosition}=\frac{f(\mathrm{Price})}{\mathrm{Volatility(Returns)}}\)$

This is very problematic:

  • Prices may live on very different scales, hence trying to find a more universal function \(f\) is almost impossible. The sign-function was a good choice as the results don’t depend on the scale of the argument.

  • Price may come with all sorts of spikes/outliers/problems.

We need a simple price filter process

  • We compute volatility-adjusted returns, filter them and compute prices from those returns.

  • Don’t call it Winsorizing in Switzerland. We apply Huber functions.

def filter(price, volatility=32, clip=4.2, min_periods=300):
    r = np.log(price).diff()
    vola = r.ewm(com=volatility, min_periods=min_periods).std()
    price_adj = (r / vola).clip(-clip, clip).cumsum()
    return price_adj

Oscillators#

  • All prices are now following a standard arithmetic Brownian motion with std \(1\).

  • What we want is the difference of two moving means (exponentially weighted) to have a constant std regardless of the two lengths.

  • An oscillator is the scaled difference of two moving averages.

def osc(prices, fast=32, slow=96, scaling=True):
    diff = prices.ewm(com=fast - 1).mean() - prices.ewm(com=slow - 1).mean()
    if scaling:
        # attention this formula is forward-looking
        s = diff.std()
        # you may want to use
        #   f,g = 1 - 1/fast, 1-1/slow
        #   s = np.sqrt(1.0 / (1 - f * f) - 2.0 / (1 - f * g) + 1.0 / (1 - g * g))
        # or a moving std
    else:
        s = 1

    return diff / s

from numpy.random import randn

price = pd.Series(data=randn(100000)).cumsum()

o = osc(price, 40, 200, scaling=True)
print("The std for the oscillator (Should be close to 1.0):")
print(np.std(o))

The std for the oscillator (Should be close to 1.0):
0.9999949999875

# from pycta.signal import osc

# take two moving averages and apply tanh
def f(price, slow=96, fast=32, vola=96, clip=3):
    # construct a fake-price, those fake-prices have homescedastic returns
    price_adj = filter(price, volatility=vola, clip=clip)
    # compute mu
    mu = np.tanh(osc(price_adj, fast=fast, slow=slow))
    return mu / price.pct_change().ewm(com=slow, min_periods=300).std()

from ipywidgets import Label, HBox, VBox, IntSlider, FloatSlider

fast = IntSlider(min=4, max=192, step=4, value=32)
slow = IntSlider(min=4, max=192, step=4, value=96)
vola = IntSlider(min=4, max=192, step=4, value=32)
winsor = FloatSlider(min=1.0, max=6.0, step=0.1, value=4.2)
left_box = VBox(
    [
        Label("Fast Moving Average"),
        Label("Slow Moving Average"),
        Label("Volatility"),
        Label("Winsorizing"),
    ]
)
right_box = VBox([fast, slow, vola, winsor])
HBox([left_box, right_box])

pos = 1e5 * f(
    prices, fast=fast.value, slow=slow.value, vola=vola.value, clip=winsor.value
)
portfolio = Portfolio.from_cashpos_prices(prices=prices, cashposition=pos, aum=1e8)

/home/runner/work/cs/cs/.venv/lib/python3.12/site-packages/pandas/core/internals/blocks.py:393: RuntimeWarning: invalid value encountered in log
  result = func(self.values, **kwargs)

portfolio.snapshot()
Contents